package com.gtja.beoms.thread;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gtja.beoms.anno.TimeAnnotation;
import com.gtja.beoms.base.CommonConstants;
import com.gtja.beoms.base.CosUrlConstants;
import com.gtja.beoms.base.FinanceConstants;
import com.gtja.beoms.base.KafkaUrlConstans;
import com.gtja.beoms.config.RedisKeyConfig;
import com.gtja.beoms.dao.*;
import com.gtja.beoms.dto.*;
import com.gtja.beoms.pojo.*;
import com.gtja.beoms.utils.*;
import com.xxl.job.core.context.XxlJobHelper;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.redisson.api.RedissonClient;
import org.redisson.client.RedisClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.transaction.annotation.Transactional;

import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @author Ghao
 * @date 2022/1/12
 */
public class MultiThreadFinance implements Runnable{
    private Logger log = LoggerFactory.getLogger(MultiThreadFinance.class);


    /** 企业微信会话存档需要的sdk */
    private String agentSecret;
    /** 一次拉取消息的条数，最大值1000，超过1000返回1000 */
    private long limit;
    /** 版本和私钥对应的map */
    private Map<Long, String> map;
    /** 公司的企业微信Id */
    private String corpId;
    /** 定时任务的参数 */
    private String[] methodParams;
    /** 公司的名称 */
    private String corpName;
    private FinanceUtils financeUtils;
    private QwContactMappingDao qwContactMappingDao;
    private QwGroupMemberDao qwGroupMemberDao;
    private QwCustomerDao qwCustomerDao;
    private QwStaffRelationDao qwStaffRelationDao;
    private QwWorkMessageDao qwWorkMessageDao;
    private ElasticSearchUtils elasticSearchUtils;
    private QwGroupDao qwGroupDao;
    private QwAccountDao qwAccountDao;
    private KafkaUrlConstans kafkaUrlConstans;
    private CosUrlConstants cosUrlConstants;
    @Value("${oms.es.databaseName}")
    private String databaseName;
    private String keyEnv;
    private RedisTemplate redisTemplate;
    private RedissonClient redissonClient;


    public MultiThreadFinance(String agentSecret, long limit, Map<Long, String> map, String corpId, String[] methodParams, String corpName, String keyEnv, RedissonClient redissonClient) {
        this.agentSecret = agentSecret;
        this.limit = limit;
        this.map = map;
        this.corpId = corpId;
        this.methodParams = methodParams;
        this.corpName = corpName;
        this.keyEnv = SpringContextUtil.getBean(RedisKeyConfig.class).getEnv();
        this.redisTemplate = SpringContextUtil.getBean("redisTemplate", RedisTemplate.class);
        this.financeUtils = SpringContextUtil.getBean(FinanceUtils.class);
        this.qwContactMappingDao = SpringContextUtil.getBean(QwContactMappingDao.class);
        this.qwGroupMemberDao = SpringContextUtil.getBean(QwGroupMemberDao.class);
        this.qwCustomerDao = SpringContextUtil.getBean(QwCustomerDao.class);
        this.qwStaffRelationDao = SpringContextUtil.getBean(QwStaffRelationDao.class);
        this.qwWorkMessageDao = SpringContextUtil.getBean(QwWorkMessageDao.class);
        this.elasticSearchUtils = SpringContextUtil.getBean(ElasticSearchUtils.class);
        this.qwGroupDao = SpringContextUtil.getBean(QwGroupDao.class);
        this.qwAccountDao = SpringContextUtil.getBean(QwAccountDao.class);
        this.redissonClient = redissonClient;
    }

    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see Thread#run()
     */
    @Override
    public void run() {
        //初始化sdk
        long sdk = financeUtils.initSDK(corpId, agentSecret);
        // sdk等于0 初期化失败
        if (sdk == 0) {
            return;
        }
        // 会话存档内容LIST
        List<JSONObject> chatDataList = new ArrayList<>();
        // 调用会话存档主程序
        //从缓存中获取消息标识
        long startSeq  = getChatDataAndChangeSeq(sdk, limit, map, chatDataList, redissonClient);

        //开始存储
        if (CollectionUtil.isNotEmpty(chatDataList)) {
            try {
                FinanceSaveData(chatDataList, corpId, startSeq, methodParams[1] , corpName);
            } catch (Exception e) {
                log.info("本次会话存档插入失败,异常信息: e[{}]", e.getMessage());
            }
        } else {
            log.info("最后一次缓存会话存档的seq: [{}]", startSeq);
        }
        //执行方法时间低于4s会有SDK崩溃导致服务down掉的风险
        ThreadUtil.safeSleep(TimeUnit.SECONDS.toMillis(1));
        financeUtils.destroySDK(sdk);
    }


    /**
     * 从腾讯拉去消息，并且修改seq
     * @return 本次拉取消息的结尾seq
     */
    private long getChatDataAndChangeSeq(long sdk, long limit, Map<Long, String> map, List<JSONObject> chatDataList, RedissonClient redissonClient) {

        return financeUtils.getChatData(sdk,corpId,keyEnv, limit, chatDataList, map, redissonClient);
    }


    public void FinanceSaveData(List<JSONObject> chatDataList, String corpId, long seq, String flag, String corpName){
        // 会话数据及会话mapping关系落库
        List<JSONObject> elasticSearchEntities = saveWeChatContactMappingAndWorkMessage(chatDataList, corpId);
        XxlJobHelper.log("本次会话存档插入数据库成功: seq[{}]", seq);
        // 会话数据存入es
//        BulkResponse bulkResponse = elasticSearchUtils.insertBatch(databaseName, elasticSearchEntities, corpId);
        // 数据写入Kafka TO 智能质检
        if (CommonConstants.DO_KAFKA_PRODUCER.equals(flag)) {
            XxlJobHelper.log("flag为[{}]执行调用kafka生产者部分", flag);
            sendWorkMessageToKafka(elasticSearchEntities, corpId, corpName, chatDataList);
        }else {
            XxlJobHelper.log("flag为[{}]不执行调用kafka生产者部分", flag);
        }
        // bulkResponse.hasFailures() false 成功 true 失败
//        if (bulkResponse == null || bulkResponse.hasFailures()) {
//            XxlJobHelper.log("本次会话存档插入ES失败: seq[{}],e[{}]", seq, bulkResponse.buildFailureMessage());
//            //自定义Exception不会回滚数据 ES出错只是让循环停止 会话数据不会回滚
//            // bulkResponse.buildFailureMessage()内有seq信息可以查看DB内的当前seq的信息分析错误原因
//            throw new RuntimeException();
//        }
        XxlJobHelper.log("本次会话存档插入ES成功: seq[{}]", seq);
    }

    @Transactional(rollbackFor = Exception.class)
    @TimeAnnotation
    public List<JSONObject> saveWeChatContactMappingAndWorkMessage(List<JSONObject> chatDataList, String corpId) {
        //es整形数据
        List<JSONObject> elasticSearchEntities = new ArrayList<>();
        for (JSONObject chatData : chatDataList) {
            XxlJobHelper.log("联系人映射关系及会话存档内容入库 chatData[{}]:", chatData.toJSONString());
            Integer seq = chatData.getInteger("seq");
            String msgId = chatData.getString("msgid");
            String action = FinanceConstants.getNameByValue(chatData.getString("action"));
            String from = chatData.getString("from");
            String roomId = chatData.getString("roomid");
            String toList = StringUtils.join(chatData.getJSONArray("tolist"), ",");
            if (StrUtil.isBlank(toList)) {
                XxlJobHelper.log("tolist为空 {}", toList);
                continue;
            }
            XxlJobHelper.log("toList不为空继续执行");
            String msgType = chatData.getString("msgtype");
            Date msgTime = new Date(chatData.getLong("msgtime"));
            String toListType;
            if (StringUtils.isEmpty(roomId) && !toList.contains(",")) {
                toListType = financeUtils.weCustomTypeJudgment(String.valueOf(toList));
            } else {
                toListType = FinanceConstants.CONTACTS_TYPE_GROUP_CHAT.getType();
            }
            String content = Optional.ofNullable(JSONObject.toJSONString(chatData.getJSONObject(msgType)))
                    .orElse(chatData.getString("content"));
            // 会话存档绑定关系入库
            QwContactMapping qwContactMapping = new QwContactMapping();
            qwContactMapping.setCorpId(corpId);
            qwContactMapping.setFromId(from);
            // 发送人类型
            qwContactMapping.setFromType(financeUtils.weCustomTypeJudgment(from));
            //接收人id和类型设定
            qwContactMapping.setReceiveId(toList);
            qwContactMapping.setReceiveType(toListType);
            qwContactMapping.setRoomId(roomId);
            Date date = new Date();
            qwContactMapping.setCreateTime(date);
            qwContactMapping.setLastUpdateTime(date);
            insertWeChatContactMapping(qwContactMapping);

            // 会话存档数据入库
            QwWorkMessage qwWorkMessage = new QwWorkMessage();
            qwWorkMessage.setCorpId(corpId);
            qwWorkMessage.setSeq(seq);
            qwWorkMessage.setMsgId(msgId);
            qwWorkMessage.setAction(action);
            qwWorkMessage.setRoomId(roomId);
            qwWorkMessage.setFromId(from);
            qwWorkMessage.setTolist(toList);
            qwWorkMessage.setTolistType(toListType);
            qwWorkMessage.setMsgType(msgType);
            qwWorkMessage.setContent(content);
            qwWorkMessage.setMsgTime(msgTime);
            qwWorkMessage.setCreateTime(date);
            qwWorkMessage.setLastUpdateTime(date);
            XxlJobHelper.log("插入企业微信工作信息表的数据", qwWorkMessage.toString());
            try {
                insertWeChatWorkMessage(qwWorkMessage);
            } catch (Exception e) {
                e.printStackTrace();
                //处理异常
                String errorSeqKey = keyEnv + "_" + corpId + "_" + "errorSeq";
                String errorMsgKey = keyEnv + "_" + corpId + "_" + "errorMsgType";
                //腾讯的逻辑是拉大于seq的，如果想要保留上次的位置，就需要用seq - 1
                Integer startSeq = chatDataList.get(chatDataList.size()).getInteger("seq") - 1;
                Integer lastSeq = chatDataList.get(0).getInteger("seq");
                String errorSeqValue =  startSeq + "_" + (lastSeq - startSeq);
                RedisUtils redisUtils = new RedisUtils();
                //存储异常的seq和size 供下次重试使用
                redisUtils.lSet(errorSeqKey, errorSeqValue, TimeUnit.DAYS.toSeconds(365));
                //存储异常类型，供发短信使用
                redisUtils.sSetAndTime(errorMsgKey, TimeUnit.DAYS.toSeconds(365), CommonConstants.CONTACT_MAPPING_TO_DB_ERROR);
            }

            // es会话存档数据
            chatData.put("corpid", corpId);
            chatData.put("msgtime", DateFormatUtils.format(msgTime, "yyyy-MM-dd HH:mm:ss"));

            //追加fromName和fromHeadUrl
            String fromName = null;
            String fromHeadUrl = null;
            switch (financeUtils.weCustomTypeJudgment(from)) {
                case CommonConstants.CONTACTS_TYPE_MEMBER:
                    Map<String, String> params = new HashMap<>();
                    params.put("corpId", corpId);
                    params.put("userQwId", from);
                    QwStaffRelation qwStaffRelation = qwStaffRelationDao.getQwStaffRelation(params);
                    if (qwStaffRelation != null) {
                        fromName = qwStaffRelation.getUserName();
                        fromHeadUrl = qwStaffRelation.getUserHeadUrl();
                    } else {
                        fromName = "未绑定的员工";
                    }
                    break;
                case CommonConstants.CONTACTS_TYPE_CUSTOMER:
                    QwCustomer qwCustomer = qwCustomerDao.getDistinctCustomInfo(corpId, from);
                    if (qwCustomer != null) {
                        // 客户表查询客户名称
                        fromName = qwCustomer.getCustName();
                        fromHeadUrl = qwCustomer.getCustHeadUrl();
                    } else if (StringUtils.isNotBlank(roomId)) {
                        // 去群聊查询微信昵称/群昵称
                        QwGroupMember qwGroupMember = qwGroupMemberDao.selectQwGroupMember(roomId, from);
                        XxlJobHelper.log("qwGroupMember为{}", qwGroupMember);
                        if (null != qwGroupMember) {
                            fromName = StringUtils.isNotBlank(qwGroupMember.getNickName()) ? qwGroupMember.getNickName() : qwGroupMember.getUserName();
                        }
                    } else {
                        fromName = "未绑定的客户";
                    }
                    break;
                case CommonConstants.CONTACTS_TYPE_ROBOT:
                    fromName = "机器人";
                    break;
                default:
            }

            chatData.put("fromName", fromName);
            chatData.put("fromHeadUrl", fromHeadUrl);
            elasticSearchEntities.add(chatData);
            XxlJobHelper.log("结束一次入库");
        }
        return elasticSearchEntities;
    }

    /**
     * 新增聊天关系映射
     *
     * @param weChatContactMapping 聊天关系映射
     * @return 结果
     */
    public void insertWeChatContactMapping(QwContactMapping weChatContactMapping) {
        List<QwContactMapping> list = qwContactMappingDao.selectQwContactMappingList(weChatContactMapping);
        if (CollectionUtil.isEmpty(list)) {
            qwContactMappingDao.insertQwContactMapping(weChatContactMapping);
        }
        return;
    }

    /**
     * 新增会话消息
     *
     * @param qwWorkMessage 会话消息
     * @return 结果
     */
    public void insertWeChatWorkMessage(QwWorkMessage qwWorkMessage) {
        QwWorkMessage param = new QwWorkMessage();
        param.setCorpId(qwWorkMessage.getCorpId());
        param.setSeq(qwWorkMessage.getSeq());
        List<QwWorkMessage> list = qwWorkMessageDao.selectQwWorkMessageList(param);
        XxlJobHelper.log("查询得到的会话存档list为{}", list.toString());
        if (CollectionUtil.isEmpty(list)) {
            try {
                qwWorkMessageDao.insertQwWorkMessage(qwWorkMessage);
                XxlJobHelper.log("插入成功");
            } catch (Exception e) {
                XxlJobHelper.log("插入失败，异常信息为：{}", e.getMessage());
            }
        }
        return;
    }

    /**
     * 会话数据写入kafka
     */
    @TimeAnnotation
    public void sendWorkMessageToKafka(List<JSONObject> chatDataList, String corpId, String corpName, List<JSONObject> dataList) {
        //初始化四个不同类型的kafka队列
        QwWorkMessaeWithToListInfoKafkaDTO textTypeList = createNewMsgTypeDTO(CommonConstants.KAFKA_MSG_TYPE_TEXT);
        QwWorkMessaeWithToListInfoKafkaDTO imageTypeList = createNewMsgTypeDTO(CommonConstants.KAFKA_MSG_TYPE_IMAGE);
        QwWorkMessaeWithToListInfoKafkaDTO videoTypeList = createNewMsgTypeDTO(CommonConstants.KAFKA_MSG_TYPE_VIDEO);
        QwWorkMessaeWithToListInfoKafkaDTO voiceTypeList = createNewMsgTypeDTO(CommonConstants.KAFKA_MSG_TYPE_VOICE);
        QwWorkMessaeWithToListInfoKafkaDTO meetingVoiceCallTypeDTO = createNewMsgTypeDTO(CommonConstants.KAFKA_MSG_TYPE_MEETING_VOICE_CALL);
        for (JSONObject chatData : chatDataList) {

            //判断toListType 1是企业员工，2是客户，3是群聊
            Integer seq = chatData.getInteger("seq");
            String msgId = chatData.getString("msgid");
            String action = FinanceConstants.getNameByValue(chatData.getString("action"));
            String from = chatData.getString("from");
            String roomId = chatData.getString("roomid");
            String toList = StringUtils.join(chatData.getJSONArray("tolist"), ",");
            String msgType = chatData.getString("msgtype");
            Date msgTime = new Date(chatData.getLong("msgtime"));
            String fromType = financeUtils.weCustomTypeJudgment(from);
            String toListType;
            if (StringUtils.isEmpty(roomId)) {
                toListType = financeUtils.weCustomTypeJudgment(String.valueOf(toList));
            } else {
                toListType = FinanceConstants.CONTACTS_TYPE_GROUP_CHAT.getType();
            }
            String content = Optional.ofNullable(JSONObject.toJSONString(chatData.getJSONObject(msgType)))
                    .orElse(chatData.getString("content"));

            // 取出formName userCode
            String fromName = null;
            String fromUserCode = null;
            //如果是顾客只查询username
            if (CommonConstants.CUSTOMER_TYPE.equals(fromType)) {
                fromName = qwCustomerDao.selectUserNameById(from, corpId);
            }
            //如果是员工还需要查询usercode
            if (CommonConstants.STAFF_TYPE.equals(fromType)) {
                StaffWorkMessageDTO staffWorkMessageDTO = qwStaffRelationDao.selectUserNameById(from, corpId);
                if (null != staffWorkMessageDTO) {
                    fromName = staffWorkMessageDTO.getUserName();
                    fromUserCode = staffWorkMessageDTO.getUserCode();
                }
            }
            // 取出tolistName userCode
            String tolistName = null;
            String tolistUserCode = null;
            //如果是顾客只查询username
            if (CommonConstants.CUSTOMER_TYPE.equals(toListType)) {
                tolistName = qwCustomerDao.selectUserNameById(toList, corpId);
            }
            //如果是员工还需要查询usercode
            if (CommonConstants.STAFF_TYPE.equals(toListType)) {
                StaffWorkMessageDTO staffWorkMessageDTO = qwStaffRelationDao.selectUserNameById(from, corpId);
                if (null != staffWorkMessageDTO) {
                    tolistName = staffWorkMessageDTO.getUserName();
                    tolistUserCode = staffWorkMessageDTO.getUserCode();
                }
            }

            // 取出群聊信息 ownerName userCode
            String owner = null;
            String ownerName = null;
            String ownerUserCode = null;
            String groupName = null;
            String notice = null;
            String status = null;
            //这个查询是根据员工来的，如果员工没有就需要到客户表中查询
            if (CommonConstants.GROUP_TYPE.equals(toListType)) {
                GroupWIthOwnerNameDTO groupWIthOwnerNameDTO = qwGroupDao.selectQwGroupwithName(roomId);
                if (null != groupWIthOwnerNameDTO) {
                    groupName = groupWIthOwnerNameDTO.getGroupName();
                    notice = groupWIthOwnerNameDTO.getNotice();
                    status = groupWIthOwnerNameDTO.getStatus();
                    if (StrUtil.isBlank(groupWIthOwnerNameDTO.getOwnerName())) {
                        String userName = qwCustomerDao.selectUserNameById(owner, corpId);
                        ownerName = userName;
                    } else {
                        ownerName = groupWIthOwnerNameDTO.getOwnerName();
                        ownerUserCode = groupWIthOwnerNameDTO.getOwnerUserCode();
                    }
                }
            }
            //组装所有的参数
            ArrayList<QwGroupMemberWithTolistInfoDTO> qwGroupMemberWithTolistInfoDTOS = qwGroupMemberDao.selectMembersByChatId(roomId);
            QwGroupWithTolistInfoDTO qwGroupWithTolistInfoDTO = new QwGroupWithTolistInfoDTO();
            qwGroupWithTolistInfoDTO.setOwnerUserCode(ownerUserCode);
            qwGroupWithTolistInfoDTO.setOwner(owner);
            qwGroupWithTolistInfoDTO.setOwnerName(ownerName);
            qwGroupWithTolistInfoDTO.setChatId(roomId);
            qwGroupWithTolistInfoDTO.setCorpId(corpId);
            qwGroupWithTolistInfoDTO.setGroupName(groupName);
            qwGroupWithTolistInfoDTO.setNotice(notice);
            qwGroupWithTolistInfoDTO.setQwGroupMembers(qwGroupMemberWithTolistInfoDTOS);
            qwGroupWithTolistInfoDTO.setStatus(status);
            QwWorkMessageWithTolistInfoDTO qwWorkMessageWithTolistInfoDTO = new QwWorkMessageWithTolistInfoDTO();
            qwWorkMessageWithTolistInfoDTO.setTolistUserCode(tolistUserCode);
            qwWorkMessageWithTolistInfoDTO.setTolistName(tolistName);
            qwWorkMessageWithTolistInfoDTO.setTolist(toList);
            qwWorkMessageWithTolistInfoDTO.setUserCode("");
            qwWorkMessageWithTolistInfoDTO.setFromName(fromName);
            qwWorkMessageWithTolistInfoDTO.setAction(action);
            qwWorkMessageWithTolistInfoDTO.setContent(content);
            qwWorkMessageWithTolistInfoDTO.setCorpId(corpId);
            qwWorkMessageWithTolistInfoDTO.setCorpName(corpName);
            qwWorkMessageWithTolistInfoDTO.setFromId(from);
            qwWorkMessageWithTolistInfoDTO.setFromType(fromType);
            qwWorkMessageWithTolistInfoDTO.setFromUserCode(fromUserCode);
            qwWorkMessageWithTolistInfoDTO.setMsgId(msgId);
            qwWorkMessageWithTolistInfoDTO.setMsgTime(msgTime);
            qwWorkMessageWithTolistInfoDTO.setMsgType(msgType);
            qwWorkMessageWithTolistInfoDTO.setQwGroupDTO(qwGroupWithTolistInfoDTO);
            qwWorkMessageWithTolistInfoDTO.setRoomId(roomId);
            qwWorkMessageWithTolistInfoDTO.setSeq(seq);
            qwWorkMessageWithTolistInfoDTO.setTolistType(toListType);
            // 判断msgType 写入四个list里面
            addQwWorkMessageWithToListInfoDTOToDiffTypeList(qwWorkMessageWithTolistInfoDTO,
                    textTypeList,
                    imageTypeList,
                    videoTypeList,
                    voiceTypeList,
                    meetingVoiceCallTypeDTO);
        }
        produceMsg(textTypeList, chatDataList);
        produceMsg(imageTypeList, chatDataList);
        produceMsg(videoTypeList, chatDataList);
        produceMsg(voiceTypeList, chatDataList);
        produceMsg(meetingVoiceCallTypeDTO, chatDataList);
    }


    private void produceMsg(QwWorkMessaeWithToListInfoKafkaDTO textTypeList, List<JSONObject> chatDataList) {
        if (null != textTypeList) {
            if (textTypeList.getCONTENT().getRECORD().size() > 0){
                String jsonString = null;
                try {
                    jsonString = JSON.toJSONString(textTypeList);
                } catch (Exception e) {
                    e.printStackTrace();
                    XxlJobHelper.log("json转换异常[{}]", e.getMessage());
                }
                HashMap<String, Object> map = new HashMap<>(8);
                map.put("key", "qwWorkMessage");
                map.put("value", jsonString);
                XxlJobHelper.log("请求生产者value是：[{}]", jsonString);
                XxlJobHelper.log("请求生产者的url是[{}]", kafkaUrlConstans.PRODUCT_TIME_NODE_URL);
                try {
                    String response = HttpRequestUtil.huToolPost(kafkaUrlConstans.PRODUCT_TIME_NODE_URL, map);
                    if (!StrUtil.contains(response,"true")){
                        //处理异常
                        String errorSeqKey = keyEnv + "_" + corpId + "_" + "errorSeq";
                        String errorMsgKey = keyEnv + "_" + corpId + "_" + "errorMsgType";
                        //腾讯的逻辑是拉大于seq的，如果想要保留上次的位置，就需要用seq - 1
                        Integer startSeq = chatDataList.get(chatDataList.size()).getInteger("seq") - 1;
                        Integer lastSeq = chatDataList.get(0).getInteger("seq");
                        String errorSeqValue =  startSeq + "_" + (lastSeq - startSeq);
                        RedisUtils redisUtils = new RedisUtils();
                        //存储异常的seq和size 供下次重试使用
                        redisUtils.lSet(errorSeqKey, errorSeqValue, TimeUnit.DAYS.toSeconds(365));
                        //存储异常类型，供发短信使用
                        redisUtils.sSetAndTime(errorMsgKey, TimeUnit.DAYS.toSeconds(365), CommonConstants.CONTACT_MAPPING_TO_DB_ERROR);
                    }
                    XxlJobHelper.log("请求生产者成功[{}]", response);
                } catch (Exception e) {
                    e.printStackTrace();
                    //处理异常
                    String errorSeqKey = keyEnv + "_" + corpId + "_" + "errorSeq";
                    String errorMsgKey = keyEnv + "_" + corpId + "_" + "errorMsgType";
                    //腾讯的逻辑是拉大于seq的，如果想要保留上次的位置，就需要用seq - 1
                    Integer startSeq = chatDataList.get(chatDataList.size()).getInteger("seq") - 1;
                    Integer lastSeq = chatDataList.get(0).getInteger("seq");
                    String errorSeqValue =  startSeq + "_" + (lastSeq - startSeq);
                    RedisUtils redisUtils = new RedisUtils();
                    //存储异常的seq和size 供下次重试使用
                    redisUtils.lSet(errorSeqKey, errorSeqValue, TimeUnit.DAYS.toSeconds(365));
                    //存储异常类型，供发短信使用
                    redisUtils.sSetAndTime(errorMsgKey, TimeUnit.DAYS.toSeconds(365), CommonConstants.CONTACT_MAPPING_TO_DB_ERROR);
                    XxlJobHelper.log("请求生产者失败：[{}]", e.getMessage());
                }
            }else {
                XxlJobHelper.log("[{}] 类型消息为空，不发送到kafka", textTypeList.getMSG_TYPE());
            }

        }
    }

    /**
     * 根据不同的msgType，将QwWorkMessageWithToListInfoDTO 放到不同类型的List中
     *
     * @param qwWorkMessageWithTolistInfoDTO 实体类
     * @param textTypeList                   文本类型的list
     * @param imageTypeList                  图像类型的list
     * @param videoTypeList                  视频类型的list
     * @param voiceTypeList                  语音类型的list
     * @param meetingVoiceCallTypeDTO
     */
    private void addQwWorkMessageWithToListInfoDTOToDiffTypeList(QwWorkMessageWithTolistInfoDTO qwWorkMessageWithTolistInfoDTO, QwWorkMessaeWithToListInfoKafkaDTO textTypeList, QwWorkMessaeWithToListInfoKafkaDTO imageTypeList, QwWorkMessaeWithToListInfoKafkaDTO videoTypeList, QwWorkMessaeWithToListInfoKafkaDTO voiceTypeList, QwWorkMessaeWithToListInfoKafkaDTO meetingVoiceCallTypeDTO) {
        String msgType = qwWorkMessageWithTolistInfoDTO.getMsgType();
        switch (msgType) {
            case CommonConstants.QW_MSG_TYPE_TEXT:
                textTypeList.getCONTENT().getRECORD().add(qwWorkMessageWithTolistInfoDTO);
                break;
            case CommonConstants.QW_MSG_TYPE_VIDEO:
                videoTypeList.getCONTENT().getRECORD().add(qwWorkMessageWithTolistInfoDTO);
                break;
            case CommonConstants.QW_MSG_TYPE_IMAGE:
                imageTypeList.getCONTENT().getRECORD().add(qwWorkMessageWithTolistInfoDTO);
                break;
            case CommonConstants.QW_MSG_TYPE_VOICE:
                voiceTypeList.getCONTENT().getRECORD().add(qwWorkMessageWithTolistInfoDTO);
                break;
            case CommonConstants.QW_MSG_TYPE_MEETING_VOICE_CALL:
                meetingVoiceCallTypeDTO.getCONTENT().getRECORD().add(qwWorkMessageWithTolistInfoDTO);
                break;
            default:
        }
    }

    /**
     * 初始化一个QwWorkMessaeWithToListInfoKafkaDTO
     *
     * @return
     */
    private QwWorkMessaeWithToListInfoKafkaDTO createNewMsgTypeDTO(String msgType) {
        QwWorkMessaeWithToListInfoKafkaDTO qwWorkMessaeWithToListInfoKafkaDTO = new QwWorkMessaeWithToListInfoKafkaDTO(msgType);
        qwWorkMessaeWithToListInfoKafkaDTO.setMSG_SUBTYPE("QW_CHAT_RECORD");
        SimpleDateFormat sd = new SimpleDateFormat("yyyyMMdd");
        SimpleDateFormat sd1 = new SimpleDateFormat("HHmmss");
        qwWorkMessaeWithToListInfoKafkaDTO.setSEND_DATE(new Integer(sd.format(new Date())));
        qwWorkMessaeWithToListInfoKafkaDTO.setSEND_TIME(new Integer(sd1.format(new Date())));
        qwWorkMessaeWithToListInfoKafkaDTO.setMSG_ID("qwWorkMessage" + RandomStringUtils.randomAlphanumeric(13));
        List<QwWorkMessageWithTolistInfoDTO> records = new ArrayList<QwWorkMessageWithTolistInfoDTO>();
        ContentDTO contentDTO = new ContentDTO();
        contentDTO.setRECORD(records);
        qwWorkMessaeWithToListInfoKafkaDTO.setCONTENT(contentDTO);
        return qwWorkMessaeWithToListInfoKafkaDTO;
    }

    /**
     * 调用cos接口发送短信
     * @param methodParams
     * @param startSeq
     * @param endSeq
     * @param produceExceptionList
     * @param qwAccountInfo
     */
    private void sendSms(String[] methodParams, long startSeq, long endSeq, HashSet<String> produceExceptionList, QwAccountInfo qwAccountInfo) throws UnsupportedEncodingException {
        //循环除了corpId的手机号
        if (methodParams.length > 2) {
            for (int i = 2; i < methodParams.length; i++) {
                StringBuffer stringBuffer = new StringBuffer();
                //加上公司名字
                if (null != qwAccountInfo) {
                    if (null != qwAccountInfo.getCorpName()) {
                        stringBuffer.append(qwAccountInfo.getCorpName());
                    }
                }
                //成功话术
                if (0 == produceExceptionList.size()) {
                    stringBuffer.append("-定时任务-会话存档执行成功，任务开始时的SEQ为")
                            .append(startSeq)
                            .append("，任务结束时的SEQ为")
                            .append(endSeq);
                } else {
                    //失败话术
                    stringBuffer.append("-定时任务-会话存档以下消息类型请求Kafka异常：")
                            .append(StringUtils.strip(produceExceptionList.toString(), "[]"));
                }
                String url = cosUrlConstants.SEND_SMS;
                String content = stringBuffer.toString();
                HashMap<String, Object> params = new HashMap<>(4);
                params.put("mobile", methodParams[i]);
                params.put("content", content);
                String response = null;
                try {
                    response = HttpRequestUtil.huToolPost(url, params);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                XxlJobHelper.log("请求cos发送短信的url为[{}]", url);
                XxlJobHelper.log("请求cos发送短信的响应结果[{}]", response);
            }
        }
    }
}

